package com.yh.csx.bsf.ods.tidb;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.Set;
import java.util.UUID;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.Header;
import com.yh.csx.bsf.core.util.LogUtils;
import com.yh.csx.bsf.core.util.PropertyUtils;
import com.yh.csx.bsf.core.util.TimeWatchUtils;
import com.yh.csx.bsf.ods.BaseCanalConsumer.Operate;
import com.yh.csx.bsf.ods.BaseCanalConsumer.OperateEnum;
import com.yh.csx.bsf.ods.StragtegyProvider;
import com.yh.csx.bsf.ods.meta.MysqlDataType;
import com.yh.csx.bsf.ods.meta.OdsColumn;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;
import com.yh.csx.bsf.ods.meta.OdsTable;
import com.yh.csx.bsf.ods.meta.OdsColumn.OdsColumnBuilder;
import com.yh.csx.bsf.ods.meta.OdsIndex;
import com.yh.csx.bsf.ods.utils.GlobalIDUtils;
import com.yh.csx.bsf.ods.utils.OdsTypeUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * Tidb 适配器
 * 
 * @author Robin.Wang
 * @date 2020-05-29
 *
 */
@Slf4j
@Service
@ConditionalOnProperty(name = "bsf.tidb.enabled", havingValue = "true")
public class TidbAdapter implements OdsSyncAdapter, InitializingBean {

	BlockingQueue<String> failedQueue = new LinkedBlockingQueue<String>();
	Executor executor = Executors.newFixedThreadPool(2);

	@Override
	public void afterPropertiesSet() throws Exception {
		executor.execute(() -> {
			while (true) {
				try {
					String sql = failedQueue.take();
					try {
						LogUtils.warn(this.getClass(), "ods", "【增量同步】SQL补偿" + sql);
						this.executeUpdate(sql);
					} catch (Exception e) {
						e.printStackTrace();
						failedQueue.offer(sql);
						LogUtils.warn(this.getClass(), "ods", "【增量同步】SQL补偿失败" + sql,e);
						Thread.sleep(1000 + new Random().nextInt(1000));
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});
	}

	@Autowired
	@Qualifier(value = "tidbDataSource")
	DataSource dataSource;

	@Autowired
	StragtegyProvider stragtegyProvider;

	/**
	 * 判断库是否存在
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createDatabase(String database) throws Exception {
		Assert.hasText(database, "table can't be null");
		String sql = "create database if not exists " + database;
		return executeUpdate(sql);
	}

	/**
	 * 判断表是否存在
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean existsTable(String table) throws Exception {
		Assert.hasText(table, "table can't be null");

		String[] dbtable = table.split("\\.");
		if (dbtable.length < 2) {
			throw new Exception("table should be {db}.{table} pattern");
		}
		String sql = "show  tables from  " + dbtable[0] + " like '" + dbtable[1] + "'";
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();
			if (ret.next()) {
				return true;
			}
			return false;
		} catch (Exception e) {
			LogUtils.error(this.getClass(), "ods", "create table exception", e);
			throw e;
		}

	}

	@Override
	public void syncStructure(OdsTable table) throws Exception {
		// 同步表结构
	}

	@Override
	public boolean createTableIfNotExists(String database, String table, List<OdsColumn> columns) throws Exception {
		if (StringUtils.hasLength(database)) {
			this.createDatabase(database);
		}

		if (!existsTable(database + "." + table)) {
			return createTable(database + "." + table, columns);
		}
		return true;
	}

	/**
	 * 创建数据表
	 * 
	 * @param table        库名.表名
	 * @param List<Column> 数据列
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createTable(String table, List<OdsColumn> columns) throws Exception {
		Set<String> ids = new HashSet<String>();
		String id = "";
		for (OdsColumn c : columns) {
			if ("ID".equals(c.getColumnName().toUpperCase())) {
				id = "id";
			}
			if ("PRI".equals(c.getColumnKey())) {
				ids.add(c.getColumnName());
			}
		}
		if (!ids.isEmpty()) {
			return createTable(table, columns, ids);
		} else if (StringUtils.hasText(id)) {
			return createTable(table, columns, new HashSet<String>() {
				{
					add("id");
				}
			});
		} else {
			return createTable(table, columns, new HashSet<String>() {
				{
					add("auto_id");
				}
			});
		}

	}

	/**
	 * 创建数据表
	 * 
	 * @param schema       数据库名
	 * @param table        表名
	 * @param List<Column> 数据列
	 * @param Set<String>  主键
	 * @return boolean
	 * @throws Exception
	 */
	public boolean createTable(String table, List<OdsColumn> columns, Set<String> ids) throws Exception {
		Assert.notEmpty(columns, "columns can't be emptry");
		StringBuilder sb = new StringBuilder();
		sb.append("create table if not exists ").append(table).append("(");
		for (OdsColumn column : columns) {
			sb.append("`").append(column.getColumnName()).append("`").append(" ");
			if ("JSON".equalsIgnoreCase(column.getDataType())) {
				sb.append(" LONGTEXT ");
			} else {
				sb.append(column.getColumnType()).append(" ");
			}

			/*
			 * if (!column.isNullable()) { sb.append(" not null "); }
			 */
			if (StringUtils.hasText(column.getColumnDefault())) {
				sb.append(" DEFAULT ").append(TidbTypeUtils.perseDefaultValue(
						MysqlDataType.valueOf(column.getDataType().toUpperCase()), column.getColumnDefault()));

			}
			if (StringUtils.hasText(column.getComment())) {
				sb.append("  COMMENT ").append("'").append(column.getComment()).append("'");
			}

			sb.append(",");
		}
		if (!sb.toString().contains("auto_id")) {
			sb.append("auto_id  bigint(64) not null default 0,");
		}
		if (!sb.toString().contains("auto_create_date")) {
			sb.append("auto_create_date DateTime default now(),");
		}
		sb.append(" PRIMARY KEY (").append(ids != null && ids.size() > 0 ? String.join(",", ids) : "auto_id")
				.append(")");
		sb.append(")").append("ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;");
		return executeUpdate(sb.toString());
	}

	public boolean createIndexs(String dbtable, List<OdsIndex> indexs) {
		String[] dbt = dbtable.split("\\.");
		return createIndexs(dbt[0], dbt[1], indexs);
	}

	public boolean createIndexs(String database, String table, List<OdsIndex> indexs) {
		return true;
	}

	@Override
	public boolean insertBatchMap(String table, List<Map<String, Object>> data, OdsTable tableStructure)
			throws Exception {

		if (data.size() < 1) {
			return false;
		}
		StringBuilder sb = new StringBuilder();
		sb.append("replace into ").append(table).append("(");
		for (OdsColumn column : tableStructure.getColumns()) {
			sb.append("`").append(column.getColumnName()).append("`").append(",");
		}
		sb.deleteCharAt(sb.length() - 1);
		sb.append(") values ");
		for (int i = 0; i < data.size(); i++) {
			sb.append("(");
			for (OdsColumn column : tableStructure.getColumns()) {
				sb.append(TidbTypeUtils.perseValue(
						MysqlDataType.valueOf(column.getDataType().equalsIgnoreCase("JSON") ? "LONGTEXT"
								: column.getDataType().toUpperCase()),
						data.get(i).get(column.getColumnName()), column.getColumnDefault())).append(",");
			}
			sb.deleteCharAt(sb.length() - 1);
			sb.append("),");
		}
		sb.deleteCharAt(sb.length() - 1);
		try {
			return this.executeUpdate(sb.toString());
		} catch (Exception e) {
			this.createTableIfNotExists(table, tableStructure.getColumns());
			return this.executeUpdate(sb.toString());
		}

	}

	/**
	 * 清空表
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 * 
	 */
	public boolean truncatTable(String table) throws Exception {
		String sql = "truncate table " + table;
		return executeUpdate(sql);
	}

	/**
	 * 删除表
	 * 
	 * @param table 库名.表名
	 * @return boolean
	 * @throws Exception
	 * 
	 */
	public boolean dropTable(String table) throws Exception {
		if (existsTable(table)) {
			String sql = "drop table if exists " + table;
			return executeUpdate(sql);
		}
		return true;
	}

	/**
	 * 重命名表名
	 * 
	 * @param oldName 原表名
	 * @param newName 新表名
	 * @return boolean
	 * @throws Exception
	 */
	public boolean renameTable(String oldName, String newName) throws Exception {
		String sql = "rename table " + oldName + " to " + newName;
		return executeUpdate(sql);
	}

	@Override
	public boolean process(List<Operate> list) throws Exception {
		for (Operate o : list) {
			String table = stragtegyProvider.getTargetTable(o.getHeader().getSchemaName(),
					o.getHeader().getTableName());
			if (StringUtils.hasText(table)) {
				this.executeUpdate(this.buildSql(o.getOperateEnum(),
						stragtegyProvider.getTargetTable(o.getHeader().getSchemaName(), o.getHeader().getTableName()),
						OdsTypeUtils.convert2OdsColumn(o.getBeforeColumns()),
						OdsTypeUtils.convert2OdsColumn(o.getAfterColumns())));
			}
		}
		return true;
	}

	public String buildSql(OperateEnum eventType, String table, List<OdsColumn> beforeColumns,
			List<OdsColumn> afterColumns) {
		switch (eventType) {
		case insert:
			return insertString(table, afterColumns);
		case update:
			return updateString(table, afterColumns);
		case delete:
			return deleteString(table, beforeColumns);
		case alterTable:
			return "";
		}
		return "";
	}

	/**
	 * 插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 */
	@Override
	public boolean insert(String table, List<OdsColumn> columns) throws Exception {
		columns.add(OdsColumnBuilder.build().setColumnName("auto_id").setDataType(MysqlDataType.BIGINT.toString())
				.setValue(GlobalIDUtils.nextId()));
		StringBuilder sb = new StringBuilder();
		sb.append("replace into ").append(table).append("(");
		for (int i = 0; i < columns.size(); i++) {
			sb.append("`").append(columns.get(i).getColumnName()).append("`").append(",");
		}
		sb.deleteCharAt(sb.length() - 1);
		sb.append(") ").append("values (");
		for (int i = 1; i <= columns.size(); i++) {
			sb.append("?,");
		}
		sb.deleteCharAt(sb.length() - 1);
		sb.append(")");
		return executeUpdate(sb.toString(), columns);
	}

	/**
	 * 插入数据
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 */
	@Override
	public void insert(Header header, List<Column> afterColumns) throws Exception {
		String table = stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName());
		if (StringUtils.hasText(table)) {
			try {
				this.insert(table, OdsTypeUtils.convert2OdsColumn(afterColumns));
			} catch (Exception e) {
				syncStrutcureAndRetry(header, null, afterColumns);
			}
		}
	}

	/**
	 * 构建插入数据SQL
	 * 
	 * @param table 库名.表名
	 * @param data  数据
	 * @return boolean
	 */
	private String insertString(String table, List<OdsColumn> columns) {
		StringBuilder sb = new StringBuilder();
		sb.append("replace into ").append(table).append("(");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(columns.get(i).getColumnName()).append(",");
		}
		sb.append("auto_id,");
		sb.append("auto_create_date");
		sb.append(") ").append("values (");
		for (int i = 0; i < columns.size(); i++) {
			sb.append(columns.get(i).getValue()).append(",");
		}
		sb.append(GlobalIDUtils.nextId()).append(",");
		sb.append("now()");
		sb.append(")");
		return sb.toString();
	}

	@Override
	public void update(Header header, List<Column> beforeColumns, List<Column> afterColumns) throws Exception {
		try {
			String table = stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName());
			if (StringUtils.hasText(table)) {
				this.executeUpdate(updateString(table, OdsTypeUtils.convert2OdsColumn(afterColumns)));
			}
		} catch (Exception e) {
			syncStrutcureAndRetry(header, null, afterColumns);
			this.executeUpdate(
					updateString(stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
							OdsTypeUtils.convert2OdsColumn(afterColumns)));

		}
	}

	/***
	 * 构建update sql
	 * 
	 * 
	 */
	public String updateString(String table, List<OdsColumn> columns) {
		StringBuilder sb = new StringBuilder();
		sb.append("update ").append(table).append(" set ");
		for (OdsColumn column : columns) {
			sb.append("`").append(column.getColumnName()).append("`").append("=").append(TidbTypeUtils
					.perseValue(MysqlDataType.valueOf(column.getDataType().toUpperCase()), column.getValue()))
					.append(",");
		}
		sb.append("`auto_id`=").append(GlobalIDUtils.nextId()).append(",");
		sb.append("`auto_create_date`=now()");
		sb.append(" where ").append(buildIdCondtion(columns));
		return sb.toString();
	}

	/**
	 * 删除数据
	 * 
	 * @param Header        canal事件头部
	 * @param beforeColumns
	 * 
	 **/
	@Override
	public void delete(Header header, List<Column> beforeColumns) throws Exception {
		String table = stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName());
		if (StringUtils.hasText(table)) {
			String sql = this.deleteString(table, OdsTypeUtils.convert2OdsColumn(beforeColumns));
			if (StringUtils.hasText(sql)) {
				try {
					this.executeUpdate(sql);
				} catch (Exception e) {
					this.failedQueue.offer(sql);
					log.error("删除事件失败", e);
					throw e;
				}
			}
		}
	}

	private String deleteString(String table, List<OdsColumn> columns) {

		String condition = buildIdCondtion(columns);
		if (!StringUtils.hasText(condition))
			return null;
		StringBuilder sb = new StringBuilder();
		sb.append("delete from ").append(table).append(" where ").append(condition);
		return sb.toString();
	}

	/**
	 * 构建主键where 条件
	 */
	private String buildIdCondtion(List<OdsColumn> afterColumns) {
		StringBuilder sb = new StringBuilder();

		for (OdsColumn column : afterColumns) {
			if ("ID".equals(column.getColumnName().toUpperCase())) {
				return " ID=" + column.getValue();
			}
			if ("PRI".equals(column.getColumnKey())) {
				sb.append("and ").append("`").append(column.getColumnName()).append("`").append("='")
						.append(column.getValue()).append("' ");

			}
		}
		if (sb.length() > 0) {
			sb.delete(0, 3);
			return sb.toString();
		}

		if (sb.length() < 1) {
			for (OdsColumn column : afterColumns) {
				sb.append("and ").append("`").append(column.getColumnName()).append("`").append("='")
						.append(column.getValue()).append("' ");
			}
			sb.delete(0, 3);
		}
		return sb.toString();
	}

	/**
	 * 查询表字段
	 * 
	 * @param table 库名.表名
	 * @return Map<String,String> 表列
	 */
	public Map<String, OdsColumn> describeTable(String table) {
		Map<String, OdsColumn> columns = new HashMap<String, OdsColumn>();
		String sql = "describe " + table;
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet ret = prepareStatement.executeQuery();
			while (ret.next()) {
				OdsColumn column = new OdsColumn();
				column.setColumnName(ret.getString("Field"));
				column.setColumnType(ret.getString("Type"));
				columns.put(column.getColumnName(), column);
			}
		} catch (Exception e) {
			log.error("Exception {}", e);
		}
		return columns;
	}

	/**
	 * 同步表结构并重试
	 */
	private void syncStrutcureAndRetry(Header header, List<Column> beforeColumns, List<Column> afterColumns)
			throws Exception {
		try {
			this.createDatabase(header.getSchemaName());
			if (!this.existsTable(stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()))) {
				this.createTable(stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
						OdsTypeUtils.convert2OdsColumn(afterColumns));
			} else {
				Map<String, OdsColumn> columns = describeTable(
						stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()));
				List<Column> newColumns = new ArrayList<Column>();
				List<Column> modifyColumns = new ArrayList<Column>();
				for (Column col : afterColumns) {
					if (!columns.containsKey(col.getName())) {
						newColumns.add(col);
					} else {

						if (col.getMysqlType().equals(columns.get(col.getName()).getColumnType()) || col.getMysqlType().equalsIgnoreCase("JSON")) {							
							continue;
						} else {
							modifyColumns.add(col);
						}
					}
				}
				if (newColumns.size() > 0 || !columns.containsKey("auto_id")
						|| !columns.containsKey("auto_create_date")) {
					this.alterTableAddColumn(
							stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
							OdsTypeUtils.convert2OdsColumn(newColumns));
					this.modifyTableAddColumn(
							stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
							OdsTypeUtils.convert2OdsColumn(modifyColumns));

				}
			}
			EventType eventType = header.getEventType();
			switch (eventType.getNumber()) {
			case EventType.INSERT_VALUE:
				this.insert(stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
						OdsTypeUtils.convert2OdsColumn(afterColumns));
				return;
			case EventType.DELETE_VALUE:
				this.delete(header, beforeColumns);
			case EventType.UPDATE_VALUE:
				this.insert(stragtegyProvider.getTargetTable(header.getSchemaName(), header.getTableName()),
						OdsTypeUtils.convert2OdsColumn(afterColumns));
			case EventType.ALTER_VALUE:
			case EventType.RENAME_VALUE:
			case EventType.CINDEX_VALUE:
			}

		} catch (Exception e) {
			throw e;
		}

	}

	/**
	 * 增加列
	 * 
	 * @param table        库名.表名
	 * @param List<Column> 增加的表列
	 * @throws Exception
	 */
	public boolean alterTableAddColumn(String table, List<OdsColumn> columns) throws Exception {

		for (OdsColumn column : columns) {
			StringBuilder sb = new StringBuilder();
			sb.append("alter table ").append(table).append(" ");
			sb.append(" add column ").append(column.getColumnName()).append(" ");
			if ("JSON".equalsIgnoreCase(column.getDataType())
					|| column.getColumnType().toLowerCase().contains("json")) {
				sb.append(" LONGTEXT ");
			} else {
				sb.append(column.getColumnType());
			}

			// if (!column.isNullable()) {
			// sb.append(" not null ");
			// }
			if (StringUtils.hasText(column.getColumnDefault())) {
				sb.append(" default ").append(column.getColumnDefault());
			}
			executeUpdate(sb.toString());

		}
		Map<String, OdsColumn> oldColumns = this.describeTable(table);
		if (!oldColumns.containsKey("auto_id")) {
			StringBuilder sb = new StringBuilder();
			sb.append("alter table ").append(table).append(" ");
			sb.append("add column auto_id bigint(64)");
			executeUpdate(sb.toString());
		}
		if (!oldColumns.containsKey("auto_create_date")) {
			StringBuilder sb = new StringBuilder();
			sb.append("alter table ").append(table).append(" ");
			sb.append("add column auto_create_date Datetime default now();");
			executeUpdate(sb.toString());

		}
		if (!oldColumns.containsKey("deleted")) {
			StringBuilder sb = new StringBuilder();
			sb.append("alter table ").append(table).append(" ");
			sb.append("add column `deleted` int default 0;");
			executeUpdate(sb.toString());

		}
		return true;
	}

	/**
	 * 修改列
	 * 
	 * @param table        库名.表名
	 * @param List<Column> 增加的表列
	 * @throws Exception
	 */
	public boolean modifyTableAddColumn(String table, List<OdsColumn> columns) throws Exception {

		for (OdsColumn column : columns) {
			StringBuilder sb = new StringBuilder();
			sb.append("alter table ").append(table).append(" ");
			sb.append(" modify column ").append(column.getColumnName()).append(" ");
			sb.append(" add column ").append(column.getColumnName()).append(" ");
			if ("JSON".equalsIgnoreCase(column.getDataType())
					|| column.getColumnType().toLowerCase().contains("json")) {
				sb.append(" LONGTEXT ");
			} else {
				sb.append(column.getColumnType());
			}
			if (StringUtils.hasText(column.getColumnDefault())) {
				sb.append(" default ").append(column.getColumnDefault());
			}
			executeUpdate(sb.toString());

		}
		return true;
	}

	@Override
	public synchronized boolean createTableIfNotExists(String dbtable, List<OdsColumn> columns) throws Exception {
		String[] dbtables = dbtable.split("\\.");
		this.createTableIfNotExists(dbtables[0], dbtables[1], columns);
		return false;
	}

	public boolean executeQuery(String sql) throws Exception {
		return (boolean) TimeWatchUtils.print(getPrintSql(), "tidb-query", () -> {
			if (StringUtils.hasText(sql)) {
				try (Connection conn = dataSource.getConnection()) {
					PreparedStatement prepareStatement = conn.prepareStatement(sql);
					prepareStatement.executeQuery();
					return true;
				} catch (Exception e) {
					LogUtils.error(this.getClass(), "ods", sql, e);
				}
			}
			return false;
		});
	}

	/**
	 * 执行查询
	 * 
	 * @throws Exception
	 */
	public boolean executeQuery(String sql, List<OdsColumn> columns) throws Exception {
		boolean ret = (boolean) TimeWatchUtils.print(getPrintSql(), "tidb-query:" + sql, () -> {
			if (StringUtils.hasText(sql)) {
				try (Connection conn = dataSource.getConnection()) {
					PreparedStatement prepareStatement = conn.prepareStatement(sql);
					for (int i = 0; i < columns.size(); i++) {
						prepareStatement.setObject(i, columns.get(i).getValue(),
								TidbTypeUtils.parseSqlType(columns.get(i).getDataType()));
					}
					prepareStatement.executeQuery();
					return true;
				} catch (Exception e) {
					LogUtils.warn(this.getClass(), "ods", "【TIDB同步】SQL执行" + sql, e);
					return false;
				}
			}
			return false;
		});
		if (!ret) {
			throw new Exception("【TIDB同步】SQL执行" + sql);
		}
		return ret;
	}

	public boolean executeUpdate(String sql) throws Exception {
		return executeUpdate(sql, true);
	}

	/**
	 * 
	 * */
	public boolean executeUpdate(String sql, boolean throwException) throws Exception {
		boolean ret = (boolean) TimeWatchUtils.print(getPrintSql(), "tidb-update:", () -> {
			LogUtils.debug(this.getClass(), "ods", "【TIDB同步】SQL执行" + sql);
			if (StringUtils.hasText(sql)) {
				try (Connection conn = dataSource.getConnection()) {
					PreparedStatement prepareStatement = conn.prepareStatement(sql);
					prepareStatement.executeUpdate();
					return true;
				} catch (Exception e) {
					LogUtils.warn(this.getClass(), "ods", "【TIDB同步】SQL执行" + sql, e);
				}
			}
			return false;
		});
		if (!ret && throwException) {
			throw new Exception("【TIDB同步】SQL执行" + sql);
		}
		return ret;

	}

	/**
	 * 执行查询
	 * 
	 * @throws Exception
	 */
	public boolean executeUpdate(String sql, List<OdsColumn> columns) throws Exception {
		boolean ret = (boolean) TimeWatchUtils.print(getPrintSql(), "tidb-update:", () -> {
			if (log.isDebugEnabled()) {
				LogUtils.debug(this.getClass(), "ods", "SQL执行:" + sql);
			}
			if (columns.isEmpty())
				return false;
			if (StringUtils.hasText(sql)) {
				try (Connection conn = dataSource.getConnection()) {
					PreparedStatement prepareStatement = conn.prepareStatement(sql);
					for (int i = 1; i <= columns.size(); i++) {
						Object value = TidbTypeUtils.perseObject(
								MysqlDataType.valueOf(columns.get(i - 1).getDataType().toUpperCase()),
								columns.get(i - 1).getValue());
						if (value == null) {
							if (columns.get(i - 1).isNullable()) {
								prepareStatement.setNull(i,
										TidbTypeUtils.parseSqlType(columns.get(i - 1).getDataType()));
							} else {
								throw new Exception("字段不为空");
							}
						} else {
							prepareStatement.setObject(i, value,
									TidbTypeUtils.parseSqlType(columns.get(i - 1).getDataType()));
						}
					}
					prepareStatement.executeUpdate();
					return true;
				} catch (Exception e) {
					LogUtils.warn(this.getClass(), "ods", "SQL执行异常" + sql + ":", e);
					LogUtils.warn(this.getClass(), "ods", "SQL执行异常数据：" + JSON.toJSONString(columns), e);
				}
			}
			return false;
		});
		if (!ret)
			throw new Exception("【全量同步】SQL执行异常" + sql);
		return ret;
	}

	/**
	 * 创建去重视图
	 */
	@Override
	public boolean createViews() throws Exception {

		return false;
	}

	@Override
	public boolean createIndex(String tableName, List<OdsIndex> indexs) throws Exception {
		HashMap<String, List<OdsIndex>> indexNames = new HashMap<String, List<OdsIndex>>();
		for (OdsIndex index : indexs) {
			if (indexNames.containsKey(index.getKeyName())) {
				indexNames.get(index.getKeyName()).add(index);
			} else {
				indexNames.put(index.getKeyName(), new ArrayList<OdsIndex>() {
					{
						add(index);
					}
				});
			}
		}
		for (Entry<String, List<OdsIndex>> item : indexNames.entrySet()) {

			StringBuilder sb = new StringBuilder();
			if ("PRIMARY".equals(item.getKey())) {

				continue;
				// sb.append(" alter table ").append(tableName);
				// sb.append(" add PRIMARY KEY ");

			} else {
				sb.append("create ");
				sb.append(item.getValue().get(0).getNonUnique() > 0 ? "" : " UNIQUE ");
				sb.append(" index ");
				if (StringUtils.hasText(item.getKey())) {
					sb.append(item.getKey() + "_idx");
				} else {
					sb.append(UUID.randomUUID().toString());
				}
				sb.append(" on ").append(tableName);
			}
			Collections.sort(item.getValue(), new Comparator<OdsIndex>() {
				@Override
				public int compare(OdsIndex o1, OdsIndex o2) {
					return o1.getSeqInIndex() - o2.getSeqInIndex();
				}
			});
			// 整理索引列
			sb.append("(");
			for (OdsIndex index : item.getValue()) {
				sb.append("`" + index.getColumnName() + "`").append(",");
			}
			sb.deleteCharAt(sb.length() - 1);
			sb.append(")");
			try {
				this.executeUpdate(sb.toString(), false);
			} catch (Exception e) {
			}

		}
		return false;
	}

	@Override
	public Integer selectCount(String sql) {
		try (Connection conn = dataSource.getConnection()) {
			PreparedStatement prepareStatement = conn.prepareStatement(sql);
			ResultSet set = prepareStatement.executeQuery();
			if (set.next()) {
				return set.getInt("count");
			}
			return 0;
		} catch (Exception e) {
			LogUtils.warn(this.getClass(), "ods", "【全量同步】SQL执行异常" + sql + ":", e);
		}
		return 0;
	}

	@Override
	public ResultSet query(String sql) throws Exception {
		return (ResultSet) TimeWatchUtils.print(getPrintSql(), "tidb-query:" + sql, () -> {
			try (Connection conn = dataSource.getConnection()) {
				PreparedStatement prepareStatement = conn.prepareStatement(sql);
				ResultSet set = prepareStatement.executeQuery();
				return set;
			} catch (Exception e) {
				LogUtils.warn(this.getClass(), "ods", "【全量同步】SQL执行异常" + sql + ":", e);
			}
			return null;
		});
	}

	private boolean getPrintSql() {
		return PropertyUtils.getPropertyCache("bsf.tidb.printSql.enabled", true);
	}

	@Override
	public void insertBatch(String dataTable, com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries)
			throws Exception {
		
		if (entries.length < 1) {
			return ;
		}
		boolean isHead = false;
		boolean hasValue = false;
		
		
		StringBuilder sb = new StringBuilder();
		sb.append("replace into " + dataTable);		
		for (int i = 0; i < entries.length; i++) {
			CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entries[i].getStoreValue());
			for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
				List<Column> columns = rowData.getAfterColumnsList();
				sb.append("(");
				if (!isHead) {
					for (Column column : columns) {
						sb.append("`").append(column.getName()).append("`").append(",");
						hasValue = true;
					}
					sb.append("`auto_id`");
					sb.append(") values (");
					isHead = true;
				}
				for (Column column : columns) {
					sb.append(TidbTypeUtils.perseValue(
							MysqlDataType.valueOf(OdsTypeUtils.getDataType(column.getMysqlType()).toUpperCase()),
							column.getValue())).append(",");
				}
				sb.append(GlobalIDUtils.nextId());
				sb.append("),");
			}
		}
		sb.deleteCharAt(sb.length() - 1);
		try {
			if (hasValue) {
				this.executeUpdate(sb.toString());
			}
		} catch (Exception e) {
			this.syncStrutcureAndRetry(entries[0].getHeader(), null, CanalEntry.RowChange
					.parseFrom(entries[0].getStoreValue()).getRowDatasList().get(0).getAfterColumnsList());
			try {
				Thread.sleep(500);
			} catch (Exception ex) {
			}
			throw e;
		}
	}

	@Override
	public void updateBatch(String dataTable, com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries)
			throws Exception {
		if (entries.length > 0) {
			StringBuilder sb = new StringBuilder();
			sb.append("replace into " + dataTable);
			boolean isHead = false;
			boolean hasVaule = false;
			for (int i = 0; i < entries.length; i++) {
				CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entries[i].getStoreValue());
				for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
					List<Column> columns = rowData.getAfterColumnsList();
					sb.append("(");
					if (!isHead) {
						for (Column column : columns) {
							sb.append("`").append(column.getName()).append("`").append(",");
							hasVaule = true;
						}
						sb.append("`auto_id`");
						sb.append(") values (");
						isHead = true;
					}
					for (Column column : columns) {
						sb.append(TidbTypeUtils.perseValue(
								MysqlDataType.valueOf(OdsTypeUtils.getDataType(column.getMysqlType()).toUpperCase()),
								column.getValue())).append(",");
					}
					sb.append(GlobalIDUtils.nextId());
					sb.append("),");
				}
			}
			sb.deleteCharAt(sb.length() - 1);
			String sql = sb.toString();
			try {
				if (hasVaule) {
					this.executeUpdate(sql);
				}
			} catch (Exception e) {
				this.syncStrutcureAndRetry(entries[0].getHeader(), null, CanalEntry.RowChange
						.parseFrom(entries[0].getStoreValue()).getRowDatasList().get(0).getAfterColumnsList());
				try {
					Thread.sleep(500);
				} catch (Exception ex) {
				}
				throw e;
			}
		}
	}

	@Override
	public void deleteBatch(String dataTable, com.alibaba.otter.canal.protocol.CanalEntry.Entry[] entries)
			throws Exception {
		if (entries != null && entries.length > 0) {
			for (int i = 0; i < entries.length; i++) {
				CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entries[i].getStoreValue());
				for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
					this.delete(entries[i].getHeader(), rowData.getBeforeColumnsList());
				}
			}

		}
	}

}
